package org.example

import org.apache.spark.sql.SparkSession

object dyfenxi {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    val filePath = "E:\\hnl\\scala09\\Scala\\src\\main\\resources\\"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
    // "ratings.dat"：UserID::MovieID::Rating::Timestamp
    // [UserID,MovieID,Rating,Timestamp]
    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {
        (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
      }
    }.cache()

//    println("平均得分最高的前 10 名的电影名称简单版")
//
//    //  (MovieId,(Rating,1)) (1200,(4.0,1))
//    rating.map(x => (x._2, (x._3.toDouble, 1)))
//      //  (MovieId,(总分,总次数))
//      .reduceByKey((x, y) => {
//        (x._1 + y._1, x._2 + y._2)
//      })
//      // (平均分,MovieId)
//      .map(x => (x._2._1 / x._2._2, x._1))
//      .sortByKey(false)
//      .take(10)
//      .foreach(println)

    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次数))")

    //MovieID::Title::Genres
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      //  (MovieId,(Rating,1)) (1252,(4.0,1))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      //( MovieId,(总分,总次数))
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    //(MovieId,(平均分,总分,总次数))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
        // (平均分,(movieId,Title,Genres,总分,总次数))
      }).sortByKey(false)
      .take(10)
      .foreach(println)

//    println("观影人数最多的前 10 部电影")
//    //(UserID,MovieID,Rating)
//    val watchViewsInfo = rating.map(x => {
//        (x._2, 1)
//      }).reduceByKey((x, y) => x + y) //  ( MovieId,总次数 )
//      .map(x => (x._2, x._1))
//      .sortByKey(false)
//      .take(10)
//    // 5 名
//    watchViewsInfo.foreach(println(_))
//    println("===================>")

//    rating.map(x => (x._2, 1)) //( MovieId, 1)
//      .reduceByKey((x, y) => {
//        (x + y)
//      }) //(MovieId,总次数)
//      .map(x => (x._2, x._1)) //  (  总次数, MovieId )
//      .sortByKey(false)
//      .take(10)
//      .foreach(println) // 286-> 999

    println("详情的输出(观影人数最多的前 10 部电影，电影编号)")
        // 输出 (总次数,(MovieID,title,Genres,总分,平均分))
        // ratingsInfo(MovieId,(平均分,总分,总次数))
        // (MovieID,((Title,Genres),(平均分,总分,总次数)))
        // ratingsInfo(MovieId,(平均分,总分,总次数))
        // moviesInfo MovieID::Title::Genres
        moviesInfo.join(ratingsInfo).map(x => {
          (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
        }).sortByKey(false)
          .take(10)
          .foreach(println)

  }
}
